#include "msg_server.h"
#include "tcp_session.h"
#include "core.h"
#include "GLog.h"
#include "file_system.h"
#include "time_helper.h"
#include <boost/lexical_cast.hpp>
#include "hy_message_dispersed_center.h"

const int MB_UNIT = 1024 * 1024;
const int KB_UNIT = 1024;

namespace hoyosvr
{
	namespace net
	{

		msg_server::msg_server(void):_is_stop(false)
		{
			net_core;
		}

		msg_server::~msg_server(void)
		{
		}

		void msg_server::init()
		{
			_client_count = 10;
			_configs = hoyosvr::file_system::load_jsonfile_val("./config/server_cfg.json");
			int port = _configs["server_port"].asInt();
			_acceptor_ptr = acceptor_pointer(new tcp::acceptor(net_core.get_io_service(), tcp::endpoint(tcp::v4(), port)));
			on_init();
			LogT << "start accept client on port --- " << color_green(port) << LogEnd;
			start_accept();
			net_core.get_logic_io_service().post(boost::bind(&msg_server::update,this));
		}

		void msg_server::run()
		{
			net_core.run();			
		}

		void msg_server::stop()
		{			
			_is_stop = true;
			on_stop();
			net_core.stop();
		}

		void msg_server::send_msg( int client_id,hy_network_message& m )
		{
			client_map::iterator it = _map_client.find(client_id);
			if ( it!=_map_client.end() )
			{		
				std::string t = m.get_string();
				hoyosvr::msg::hoyo_msg::ptr hm = hoyosvr::msg::hoyo_msg::create( t.c_str(), t.length() );
				it->second->write_hoyo_msg( *hm );
			}
		}

		void msg_server::send_msg( tcp_session_ptr conn_ptr,hy_network_message* m )
		{
			std::string t = m->get_string();
			hoyosvr::msg::hoyo_msg::ptr hm = hoyosvr::msg::hoyo_msg::create( t.c_str(), t.length() );
			conn_ptr->write_hoyo_msg(*hm);
		}

		void msg_server::disconnect( int client_id )
		{

		}

		void msg_server::on_connect( int client_id,tcp_session_ptr conn_ptr )
		{
			
		}

		void msg_server::on_disconnect( int client_id,tcp_session_ptr conn_ptr )
		{
			boost::mutex::scoped_lock lock(_client_mutex);

			client_map::iterator i = _map_client.find(client_id);
			if(i != _map_client.end())
			{
				_map_client.erase(i);
				LogT <<  "client[" << client_id << "] disconnected ..." << LogEnd;
			}
		}

		void msg_server::sync_process_recv_msg( int net_id,hy_network_message::ptr recv_msg_ptr )
		{
		}

		bool msg_server::async_process_recv_msg(tcp_session_ptr conn,hy_network_message::ptr recv_msg_ptr)
		{
			return false;
		}

		void msg_server::on_update()
		{
		}

		void msg_server::start_accept()
		{			
			size_t n = _configs["buffer_size"].asUInt(); //byte 
			n *= 1024;//K
			n *= 1024;//M
			tcp_session_ptr new_connection = tcp_session::create( n ); // n(M)

			_acceptor_ptr->async_accept(new_connection->socket(),
				boost::bind(&msg_server::handle_accept,this, new_connection,_acceptor_ptr,
				boost::asio::placeholders::error));
		}

		void msg_server::handle_accept(tcp_session_ptr session,acceptor_pointer ap,const boost::system::error_code& error)
		{
			if (!error)
			{
				{
					boost::mutex::scoped_lock lock(_client_mutex);
					_map_client[++_client_count] = session;
				}
				try
				{
					std::string  ip_add = session->socket().remote_endpoint().address().to_string();
					LogI <<  "new client[" << _client_count << "] connected ...ip: " << ip_add << LogEnd;
				}
				catch (std::exception& e)
				{
					LogE << "handler accept error: " << e.what() << LogEnd;
				}
				session->start(_client_count,shared_from_this());

				start_accept();
			}
			else
			{
				try
				{
					session->stop();
					std::string  ip_add = session->socket().remote_endpoint().address().to_string();
					LogE <<  __FUNCTION__ << error.message() <<  "\tip:" << ip_add.c_str() << LogEnd;
				}
				catch (std::exception& e)
				{
					LogE << "error: " << e.what() << LogEnd;
				}
			}
		}

		void msg_server::recv_client_handler( tcp_session_ptr session,const char* data_ptr,int len )
		{
			if ( len < hoyosvr::msg::msg_offset ) return;

			hy_network_message::ptr p = hy_network_message::create(data_ptr,len);

			boost::mutex::scoped_lock lock(_queue_mutex);
			if ( !async_process_recv_msg(session, p ) )
			{
				msg_bind msg;
				msg.msg = p;
				msg.net_id = session->get_net_id();
				_msg_queue.push(msg);		
			}
		}

		// should disconnect from client in case
		void msg_server::client_connect_handler( tcp_session_ptr session,int error_value )
		{
			on_disconnect(session->get_net_id(),session);
		}

		void msg_server::on_init()
		{

		}

		void msg_server::update()
		{
			if(_is_stop) 
			{				
				return;
			}
			static boost::system_time start_time;
			start_time = boost::get_system_time();
			
			// main loop
			while ( 1)
			{
				// lock lock lock
				boost::mutex::scoped_lock lock(_queue_mutex);
				if (!_msg_queue.empty())
				{				
					int net_id = _msg_queue.front().net_id;
					hy_network_message::ptr p = _msg_queue.front().msg;
					_msg_queue.pop();

					// unlock right now
					lock.unlock();

					// process the message now
					boost::system_time startTime = boost::get_system_time();
					sync_process_recv_msg( net_id, p );
					boost::system_time endTime = boost::get_system_time();
					int lapseTime = (int)((endTime - startTime).total_milliseconds());
					if ( lapseTime > 1 )
					{
						string t = boost::lexical_cast<string>(lapseTime);
						LogD << "client: " << p->get_player_id() << " cmd [" << p->get_protocol_id()
							<< "] consume time : " << color_red(t) << LogEnd;
					}
				}
				else
					break;
				
				// don't spend too much time
				boost::system_time end_time = boost::get_system_time();
				if((end_time - start_time).total_milliseconds() >= 100)
				{
					break;
				}
			}			
			
			on_update();

			// sleep 1 micro second
			hoyosvr::time_helper::sleepMicroSec(1);

			net_core.get_logic_io_service().post(boost::bind(&msg_server::update,this));
		}

	}
}

